package com.yanggu.hotitems_analysis

import cn.hutool.core.collection.BoundedPriorityQueue
import cn.hutool.core.date.DateUtil
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date
import java.util.concurrent.TimeUnit
import scala.collection.convert.ImplicitConversions.`list asScalaBuffer`

/**
 * 统计热门商品topN
 */
object HotItems {

  def main(args: Array[String]): Unit = {

    //创建流式执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //为了测试方便, 设置并行度为1
    env.setParallelism(1)
    //高版本的flink（1.12开始）默认就是事件时间
    //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val kafkaSource = KafkaSource.builder[String]()
      .setBootstrapServers("localhost:9092")
      .setTopics("hotitems")
      .setGroupId("hotitems")
      .setStartingOffsets(OffsetsInitializer.earliest())
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build()

    env
      //从文件中读取数据
      //.readTextFile(getClass.getResource("/UserBehavior.csv").getFile)
      //从Kafka中读取数据
      .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "hotitems")
      //转换成样例类, 并且只关注pv
      .flatMap((dataLine, out: Collector[UserBehavior]) => {
        val arr = dataLine.split(",")
        //这里只关注pv
        if (arr(3) == "pv") {
          out.collect(UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, "pv", arr(4).toLong))
        }
      })
      //设置EventTime, 这里不设置Watermark, 自己决定窗口的数据什么时候进行计算
      .assignAscendingTimestamps(_.timestamp * 1000)
      //按照itemId进行分组 dataStream => KeyedStream
      .keyBy(_.itemId)
      //设置一个事件时间滑动窗口, 窗口大小1小时, 滑动步长5分钟
      .window(SlidingEventTimeWindows.of(Time.hours(1L), Time.minutes(5L)))
      //使用aggregate方法, 这里需要知道窗口的开始时间和结束时间
      .aggregate(new ItemCountAggregateFunction, new ItemViewCountWindowFunction)
      //根据窗口开始和结束时间进行分组 dataStream => KeyedStream
      .keyBy(data => (data.windowStart, data.windowEnd))
      //这里由于进行了分组, 需要使用KeyedProcessFunction
      .process(new ItemViewCountProcessFunction)
      //进行打印输出
      .print("HotItems Analysis")

    env.execute("HotItems Job")

  }

}

/**
 * 用户行为的实体类。输入数据的样例类
 *
 * @param userId     用户id
 * @param itemId     商品id
 * @param categoryId 商品的分类id
 * @param behavior   具体的行为
 * @param timestamp  时间戳。这里是秒，不是毫秒
 */
case class UserBehavior(userId: Long, itemId: Long, categoryId: Int, behavior: String, timestamp: Long)

/**
 * 定义窗口聚合结果样例类
 *
 * @param itemId      商品id
 * @param windowStart 窗口开始时间
 * @param windowEnd   窗口结束时间
 * @param count       访问数量
 */
case class ItemViewCount(itemId: Long, windowStart: Long, windowEnd: Long, count: Long)

/**
 * COUNT 统计的聚合函数实现，每出现一条记录加一
 * 进行预聚合, 提前进行累加运算
 *
 * @tparam IN  UserBehavior 输入的数据类型
 * @tparam ACC Long 累加的中间状态
 * @tparam OUT Long 输出的数据类型
 */
class ItemCountAggregateFunction extends AggregateFunction[UserBehavior, Long, Long] {

  override def createAccumulator(): Long = 0L

  override def add(value: UserBehavior, accumulator: Long): Long = accumulator + 1L

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b

}

/**
 * 用于输出窗口的结果
 * WindowFunction将每个Key每个窗口聚合后的结果带上其他信息进行输出
 *
 * @tparam IN  Long 累加器输出的数据类型作为输入值的类型
 * @tparam OUT ItemViewCount 输出的数据类型
 * @tparam KEY Long 分组的key的数据类型
 * @tparam W   TimeWindow 窗口的类型, 一般是时间窗口
 */
class ItemViewCountWindowFunction extends ProcessWindowFunction[Long, ItemViewCount, Long, TimeWindow] {

  override def process(key: Long, context: Context, elements: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
    out.collect(ItemViewCount(key, context.window.getStart, context.window.getEnd, elements.head))
  }

}

/**
 * 求某个窗口中前 N 名的热门点击商品，key 为窗口开始时间和结束时间戳，输出为 TopN 的结果字符串
 * 正常可以定义Sink, 输出到其他系统
 */
class ItemViewCountProcessFunction extends KeyedProcessFunction[(Long, Long), ItemViewCount, String] {

  //设置ValueState用于存储窗口内的数据, 泛型是一个有界优先队列
  private lazy val valueState: ValueState[BoundedPriorityQueue[ItemViewCount]] = getRuntimeContext.getState(
    new ValueStateDescriptor[BoundedPriorityQueue[ItemViewCount]]("valueState", classOf[BoundedPriorityQueue[ItemViewCount]]))

  override def processElement(value: ItemViewCount,
                              ctx: KeyedProcessFunction[(Long, Long), ItemViewCount, String]#Context,
                              out: Collector[String]): Unit = {

    //这里不能在open方法中初始化有界有限队列, 只能在processElement方法中初始化
    var counts = valueState.value()
    //如果为空初始化一个有界优先队列
    if (counts == null) {
      //创建一个有界优先队列, 大小为10
      counts = new BoundedPriorityQueue[ItemViewCount](10, (o1: ItemViewCount, o2: ItemViewCount) => o2.count.compare(o1.count))
      valueState.update(counts)
    }

    //把数据添加到有界优先队列中
    counts.offer(value)

    //设置定时器, 在窗口的结束时间 + 100执行定时器, 在窗口的结束时间 + 100时刻认为所有的数据都已经收集完成, 输出窗口数据
    //可以简单的理解为这里设置一个Watermark, 大小为窗口结束时间 + 100毫秒, 到时候进行窗口输出
    ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey._2 + 100)
  }

  //在定时器中进行统计输出窗口数据
  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[(Long, Long), ItemViewCount, String]#OnTimerContext,
                       out: Collector[String]): Unit = {
    val window = ctx.getCurrentKey
    val counts = valueState.value()
    val buffer = counts.toList

    //清空状态, 这里只是清除引用
    valueState.clear()

    val builder = new StringBuilder

    builder.append(s"窗口开始时间: ${DateUtil.formatDateTime(new Date(window._1))}, " +
      s"时间窗口: ${DateUtil.formatDateTime(new Date(window._2))}").append("\r\n")

    for (index <- buffer.indices) {
      val data = buffer(index)
      builder.append(s"排名: ${index + 1}, 商品id: ${data.itemId}, 访问次数: ${data.count}").append("\r\n")
    }

    //收集数据进行输出
    out.collect(builder.toString())

    //阻塞线程1秒, 防止输出过快, 模拟数据
    TimeUnit.SECONDS.sleep(1L)
  }

}